use std::sync::Arc;
use std::time::Duration;
use anyhow::Result;
use serde::{Deserialize, Serialize};
use tracing::{info, Level};
use tracing_subscriber::FmtSubscriber;
use async_trait::async_trait;

use protoactor_rs::actor::{Actor, ActorContext, ActorError, DefaultActorContext, PID};
use protoactor_rs::persistence::{
    PersistentActor, PersistentEvent, PersistentState, PersistenceError, 
    PostgresEventStore, EventStore, PersistentActorContext
};
use protoactor_rs::system::ActorSystem;

// 事件定义
#[derive(Debug, Clone, Serialize, Deserialize)]
enum CounterEvent {
    Incremented(i32),
    Decremented(i32),
    Reset,
}

impl PersistentEvent for CounterEvent {
    fn event_type(&self) -> &str {
        match self {
            CounterEvent::Incremented(_) => "incremented",
            CounterEvent::Decremented(_) => "decremented",
            CounterEvent::Reset => "reset",
        }
    }
}

// 状态定义
#[derive(Debug, Clone, Serialize, Deserialize)]
struct CounterState {
    count: i32,
    version: u64,
}

impl PersistentState for CounterState {
    fn version(&self) -> u64 {
        self.version
    }
    
    fn set_version(&mut self, version: u64) {
        self.version = version;
    }
}

// 持久化Actor
struct CounterActor {
    count: i32,
}

impl CounterActor {
    fn new() -> Self {
        Self { count: 0 }
    }
}

#[async_trait]
impl Actor for CounterActor {
    type Context = PersistentActorContext<DefaultActorContext, CounterEvent, CounterState>;
    
    async fn receive(&self, _ctx: &mut Self::Context, _msg: Box<dyn std::any::Any + Send>) -> Result<(), ActorError> {
        // 具体消息处理逻辑由PersistentActor的handle_message实现
        Ok(())
    }
}

#[async_trait]
impl PersistentActor for CounterActor {
    type Event = CounterEvent;
    type State = CounterState;
    type Context = PersistentActorContext<DefaultActorContext, CounterEvent, CounterState>;
    
    fn apply_event(&mut self, event: &Self::Event) {
        match event {
            CounterEvent::Incremented(value) => {
                self.count += value;
                info!("Applied Increment: count = {}", self.count);
            }
            CounterEvent::Decremented(value) => {
                self.count -= value;
                info!("Applied Decrement: count = {}", self.count);
            }
            CounterEvent::Reset => {
                self.count = 0;
                info!("Applied Reset: count = {}", self.count);
            }
        }
    }
    
    fn apply_snapshot(&mut self, snapshot: &Self::State) {
        self.count = snapshot.count;
        info!("Applied Snapshot: count = {}", self.count);
    }
    
    fn get_state(&self) -> Self::State {
        CounterState {
            count: self.count,
            version: 0, // 这个会在保存时被设置
        }
    }
    
    async fn handle_message(&mut self, context: &mut Self::Context, message: Box<dyn std::any::Any + Send>) -> Result<(), ActorError> {
        if let Some(cmd) = message.downcast_ref::<&str>() {
            match *cmd {
                "increment" => {
                    let event = CounterEvent::Incremented(1);
                    context.persist(event).await?;
                    context.respond(self.count);
                }
                "increment_by" => {
                    if let Some(value) = message.downcast_ref::<i32>() {
                        let event = CounterEvent::Incremented(*value);
                        context.persist(event).await?;
                        context.respond(self.count);
                    }
                }
                "decrement" => {
                    let event = CounterEvent::Decremented(1);
                    context.persist(event).await?;
                    context.respond(self.count);
                }
                "reset" => {
                    let event = CounterEvent::Reset;
                    context.persist(event).await?;
                    context.respond(self.count);
                }
                "snapshot" => {
                    let state = self.get_state();
                    context.save_snapshot(&state).await?;
                    context.respond("Snapshot saved");
                }
                "get" => {
                    context.respond(self.count);
                }
                _ => {
                    context.respond("Unknown command");
                }
            }
        }
        Ok(())
    }
    
    async fn on_recovery_complete(&mut self, _context: &mut Self::Context) -> Result<(), ActorError> {
        info!("Recovery complete: count = {}", self.count);
        Ok(())
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // 设置日志
    let subscriber = FmtSubscriber::builder()
        .with_max_level(Level::INFO)
        .finish();
    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");
    
    // 创建PostgreSQL存储
    let postgres_store = PostgresEventStore::<CounterEvent, CounterState>::new(
        "postgres://postgres:postgres@localhost/protoactor",
        "proto",
        "events",
        "snapshots",
        10, // 每10个事件保存一次快照
    ).await?;
    
    // 创建Actor系统
    let system = ActorSystem::new();
    
    // 创建持久化Actor
    let counter_actor = CounterActor::new();
    let counter_pid = protoactor_rs::persistence::spawn_persistent_actor(
        &system,
        counter_actor,
        Arc::new(postgres_store),
        "counter-1".to_string(),
    ).await?;
    
    // 使用Actor
    for i in 0..5 {
        let response = system.request::<i32>(&counter_pid, "increment", Duration::from_secs(1)).await?;
        info!("Counter value after increment {}: {}", i+1, response);
    }
    
    // 重置计数器
    let response = system.request::<i32>(&counter_pid, "reset", Duration::from_secs(1)).await?;
    info!("Counter value after reset: {}", response);
    
    // 增加计数器
    let response = system.request::<i32>(&counter_pid, "increment", Duration::from_secs(1)).await?;
    info!("Counter value after increment: {}", response);
    
    // 保存快照
    let response = system.request::<String>(&counter_pid, "snapshot", Duration::from_secs(1)).await?;
    info!("Snapshot response: {}", response);
    
    // 允许所有操作完成
    tokio::time::sleep(Duration::from_secs(1)).await;
    
    Ok(())
} 